package org.KeyCrafter.server;

import cn.hutool.core.io.FileUtil;
import com.esotericsoftware.kryo.Kryo;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufAllocator;
import org.KeyCrafter.cmd.Command;
import org.KeyCrafter.ds.dict.*;
import org.KeyCrafter.obj.*;
import org.KeyCrafter.obj.type.DefaultStringType;
import org.KeyCrafter.obj.type.HashType;
import org.KeyCrafter.obj.type.ListType;
import org.KeyCrafter.obj.type.SetType;
import org.KeyCrafter.server.aof.AofAttribute;
import org.KeyCrafter.server.aof.LoadingInfo;
import org.KeyCrafter.server.session.ClientSession;
import org.KeyCrafter.util.ProgressBar;

import java.io.*;
import java.nio.channels.FileChannel;
import java.nio.charset.StandardCharsets;
import java.nio.file.*;
import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.logging.Level;
import java.util.logging.Logger;

import static org.KeyCrafter.obj.KeyCrafterObject.*;
import static org.KeyCrafter.server.BackgroundServer.KeyCrafter_BS_AOF_FSYNC;
import static org.KeyCrafter.server.ServerConstants.*;
import static org.KeyCrafter.server.aof.AofAttribute.KeyCrafter_AOF_OFF;
import static org.KeyCrafter.server.aof.AofAttribute.KeyCrafter_AOF_ON;
import static org.KeyCrafter.util.CodecUtil.encodeArray;
import static org.KeyCrafter.util.RewriteUtil.writeArray;
import static org.KeyCrafter.util.TimeUtil.microTime;

public class AppendOnlyFile {
    public static final int AOF_FSYNC_NO = 0;
    public static final int AOF_FSYNC_ALWAYS = 1;
    public static final int AOF_FSYNC_EVERYSEC = 2;
    // 错误日志频率（每秒）
    public static final int AOF_WRITE_LOG_ERROR_RATE = 30;
    private static final ServerContext sc = ServerContext.DEFAULT;
    // 上次修改日志的时间
    private static long lastWriteErrorLog = 0;
    private static final Logger logger = Logger.getLogger(AofAttribute.class.getName());
    // aof 重写线程池
    private static final ExecutorService aofRewriteExecutor = Executors.newSingleThreadExecutor();

    // 将 AOF 缓存写入到文件中。
    public static void flushAppendOnlyFile(boolean force) {
        boolean  syncInProgress = false; // 后台是否有同步程序
        // 缓冲区中没有任何内容或者不可读，直接返回
        AofAttribute aofAttr = sc.getAOFAttribute();
        ByteBuf aofBuf = aofAttr.getBuf();
        if (aofBuf == null || !aofBuf.isReadable()) {
            return;
        }
        BackgroundServer bg = sc.getBackgroundServer();
        // 策略为每秒 FSYNC
        if (aofAttr.getFsync() == AOF_FSYNC_EVERYSEC) {
            // 是否有 SYNC 正在后台进行？
            syncInProgress = bg.pendingJobsOfType(KeyCrafter_BS_AOF_FSYNC) != 0;
        }
        // 每秒 fsync ，并且强制写入为假
        if (aofAttr.getFsync() == AOF_FSYNC_EVERYSEC && !force) {
            /*
             * 当 fsync 策略为每秒钟一次时， fsync 在后台执行。
             *
             * 如果后台仍在执行 FSYNC ，那么我们可以延迟写操作一两秒
             * （如果强制执行 write 的话，服务器主线程将阻塞在 write 上面）
             */
            if (syncInProgress) {
                if (aofAttr.getFlushPostponedStart() == 0) {
                    /* 前面没有推迟过 write 操作，这里将推迟写操作的时间记录下来
                     * 然后就返回，不执行 write 或者 fsync
                     * （Reids精确到微秒，这里先用毫秒）
                     */
                    aofAttr.setFlushPostponedStart(sc.mstime());
                    return;
                } else if (sc.mstime() - aofAttr.getFlushPostponedStart() < 2000) {
                    /*如果之前已经因为 fsync 而推迟了 write 操作
                     * 但是推迟的时间不超过 2 秒，那么直接返回
                     * 不执行 write 或者 fsync
                     */
                    return;
                }
                /* Otherwise fall trough, and go write since we can't wait
                 * over two seconds.
                 *
                 * 如果后台还有 fsync 在执行，并且 write 已经推迟 >= 2 秒
                 * 那么执行写操作（write 将被阻塞）
                 */
                aofAttr.incrDelayedFsync(); // 操作推迟数加一
                logger.log(Level.INFO,"Asynchronous AOF fsync is taking too long (disk is busy?). Writing the AOF buffer without waiting for fsync to complete, this may slow down KeyCrafter.");

            }
        }

        /* 对 AOF 文件进行写入。
         * 清零延迟 write 的时间记录
         */
        aofAttr.setFlushPostponedStart(0);
        /* 执行单个 write 操作，如果写入设备是物理的话，那么这个操作应该是原子的
         * 当然，如果出现像电源中断这样的不可抗现象，那么 AOF 文件也是可能会出现问题的
         */
        FileChannel aofOut = aofAttr.getChannel();
        int nwritten = 0;
        try {
            // 返回ByteBuf中可以读取的字节的数量。在写入数据后读取数据之前
            int expected = aofBuf.readableBytes();
            // 返回值等于写入的字节的数量。当你读取数据时，readableBytes()的值会减少。
            nwritten = aofBuf.readBytes(aofOut, expected);

            if (nwritten != expected){
                boolean canLog = false;
                // 控制日志输出频率
                // unit = ms
                if ((sc.mstime() - lastWriteErrorLog) > AOF_WRITE_LOG_ERROR_RATE * 1000) {
                    canLog =true;
                    lastWriteErrorLog = sc.mstime();
                }
                if (canLog){
                    // 预期字节量少于输入字节量
                    logger.log(Level.WARNING,String.format("Short write while writing to the AOF" +
                            "file:(nwritten=%d,expected=%d)", nwritten, expected));
                }
                // 尝试移除新追加的不完整内容
                try {
                    // 可能是是磁盘不足了？
                    aofOut.truncate(aofAttr.getCurrentSize());
                    // 成功调整大小
                    nwritten = -1;
                }catch (IOException ex) {
                    // 调整大小失败
                    if (canLog) {
                        logger.log(Level.WARNING,String.format("Could not remove short write " +
                                "from the append-only file.  KeyCrafter may refuse " +
                                "to load the AOF the next time it starts.  " +
                                "ftruncate: %s", ex.getMessage()));
                    }
                    throw ex; // 抛到外层去处理
                }
                // 写入成功，更新最后写入状态
                if (aofAttr.getLastWriteStatus() == KeyCrafter_ERR) {
                    logger.log(Level.WARNING,"AOF write error looks solved," +
                            " KeyCrafter can write again.");
                    aofAttr.setLastWriteStatus(KeyCrafter_OK);
                }
            }
        } catch (IOException e) {
            boolean canLog = false;
            if ((sc.mstime() - lastWriteErrorLog) > AOF_WRITE_LOG_ERROR_RATE*1000) {
                canLog = true;
                lastWriteErrorLog = sc.mstime(); // 时间精度问题
            }
            // 如果写入出错，那么尝试将该情况写入到日志里面
            if (canLog) {
                logger.log(Level.WARNING,String.format("Error writing to the AOF file: %s",e.getMessage()));
                aofAttr.setLastWriteEx(e);
            }
            // 处理写入 AOF 文件时出现的错误
            if (aofAttr.getFsync() == AOF_FSYNC_ALWAYS) {
                logger.log(Level.WARNING,"Can't recover from AOF write error when the AOF fsync policy is 'always'. Exiting...");
                System.exit(1);
            } else {
                aofAttr.setLastWriteStatus(KeyCrafter_ERR);
                if (nwritten > 0) {
                    aofAttr.setCurrentSize(aofAttr.getCurrentSize()+nwritten);
                    // ByteBuf 会调整读指针的到已读位置，所以这里不用调整 aofBuf （大概，待考证。。。）
                }
                return;
            }
        }
        // 更新写入后的 AOF 文件大小
        aofAttr.setCurrentSize(aofAttr.getCurrentSize()+nwritten);
        // 清理缓存
        aofAttr.getBuf().clear();
        // fsnyc 策略
        // 如果是 AOF_FSYNC_ALWAYS，已经在 FileChannel 中启动了 DSYNC 操作
        // DSYNC 不会更新元数据，Redis也是如此
        // 如果是 AOF_FSYNC_NO，则不需要处理，由操作系统决定何时同步
        // 所以只需要考虑 AOF_FSYNC_EVERYSEC 的情况

        // 策略为每秒 fsnyc ，并且距离上次 fsync 已经超过 1 秒
        // (先用毫秒表示，后期可以考虑用Java的时间类以获得微秒时间戳)
        if ((aofAttr.getFsync() == AOF_FSYNC_EVERYSEC &&
                sc.mstime() - aofAttr.getLastFsync() > 1000)) {
            // 放到后台执行
            if (!syncInProgress) {
                aofBackgroundFsync(aofAttr.getChannel());
            }
            // 更新最后一次执行 fsnyc 的时间 （Redis是微秒，这里还是先用毫秒）
            aofAttr.setLastFsync(sc.mstime());
        }
    }

    /*
     * 在另一个线程中，对给定 channel 执行一个后台 fsync() 操作。
     */
    private static void aofBackgroundFsync(FileChannel channel) {
        sc.getBackgroundServer().createBackgroundJob(KeyCrafter_BS_AOF_FSYNC,channel);
    }

    /*
     * 将命令追加到 AOF 文件中，
     * 如果 AOF 重写正在进行，那么也将命令追加到 AOF 重写缓存中。
     */
    public static void feedAppendOnlyFile(Command cmd, int dbId,  ByteBuf queryBuf){
        AofAttribute aofAttr = sc.getAOFAttribute();
        ByteBuf buf = ByteBufAllocator.DEFAULT.buffer();
        // 使用 SELECT 命令，显式设置数据库，确保之后的命令被设置到正确的数据库
        if (dbId != aofAttr.getSelectedDb()) {
            String seldb = String.valueOf(dbId);
            String dbCmd = String.format("*2\r\n$6\r\nSELECT\r\n$%d\r\n%s\r\n", seldb.length(), seldb);
            aofAttr.setSelectedDb(dbId);
            buf.writeBytes(dbCmd.getBytes(StandardCharsets.UTF_8));
        }
        // 因为前面经过解码，读指针和写指针重合了，不能直接写入queryBuf
        // 显式的指定写入的起始位置为 0，写入长度为写指针位置（写指针指向的时下一个将要写的位置）
        buf.writeBytes(queryBuf, 0, queryBuf.writerIndex());
        // EXPIRE 、 PEXPIRE 和 EXPIREAT 命令(先略过)
        // SETEX 和 PSETEX 命令 (先略过)

        // 先考虑普通命令

        /* 将命令追加到 AOF 缓存中，
         * 在重新进入事件循环之前，这些命令会被冲洗到磁盘上，
         * 并向客户端返回一个回复。
         */
        if (aofAttr.getState() == KeyCrafter_AOF_ON) {
            aofAttr.getBuf().writeBytes(buf,buf.readableBytes());
        }

        /* 如果 BGREWRITEAOF 正在进行，
         * 那么我们还需要将命令追加到重写缓存中，
         * 从而记录当前正在重写的 AOF 文件和数据库当前状态的差异。
         */
        if (aofAttr.getRewriteThreadId() != -1) {
            // buf 被读过了，
            aofAttr.getRewriteBuf().writeBytes(buf,0,buf.writerIndex());
        }
        // 释放buf
        buf.release();
    }
    /*  执行 AOF 文件中的命令。
     * 出错时返回 REDIS_OK 。
     * 出现非执行错误（比如文件长度为 0 ）时返回 REDIS_ERR 。
     * 出现致命错误时打印信息到日志，并且程序退出。
     */
    public static int loadAppendOnlyFile(String filename){
        AofAttribute aofAttr = sc.getAOFAttribute();
        // 伪客户端
        ClientSession fakeClient = new ClientSession();
        // 校验文件的正确性
        Path path = Paths.get(filename);
        File file = path.toFile();
        if (!file.exists() || file.length() == 0) {
            aofAttr.setCurrentSize(0);
            return KeyCrafter_ERR;
        }
        // 校验文件的后缀
        String suffix = FileUtil.getSuffix(filename);
        final List<String> validFileSuffixList = Arrays.asList("aof");
        if (!(validFileSuffixList.contains(suffix))){
            return KeyCrafter_ERR;
        }
        // 打开 AOF 文件
        RandomAccessFile br = null;
        try {
            br = new RandomAccessFile(file,"r");
        } catch (FileNotFoundException e) {
            logger.log(Level.WARNING, String.format("Fatal error: can't open the append log file for reading: %s", e.getMessage()));
            System.exit(1);
        }
        // 暂时性地关闭 AOF
        int oldAofState = aofAttr.getState();
        aofAttr.setState(KeyCrafter_AOF_OFF);
        // 设置服务器的状态为：正在载入
        LoadingInfo loadingInfo = sc.getLoadingInfo();
        loadingInfo.startLoading(file);

        int loops = 0;
        // 已读行
        int readLines = 0;
        // 进度条
        ProgressBar progressBar = new ProgressBar(loadingInfo.getTotalBytes(), 50, '█');
        // 开始加载aof
        logger.log(Level.INFO,"DB started loading append only file");
        while (true) {
            if ((loops++ % 1000) == 0) {
                // 刷新加载进度
                try {
                    loadingInfo.loadingProgress(br.getFilePointer());
                    progressBar.show(loadingInfo.getLoadedBytes());
                } catch (IOException e) {
                    logReadErrAndExit(e);
                }
            }
            // 读入一行文件内容到缓存
            char[] buf = null;
            String line;
            try {
                line = br.readLine();
                readLines++;
                if (line == null) {
                    // 文件已经读完，跳出
                    break;
                }
                buf = line.toCharArray();
            } catch (IOException e) {
                logReadErrAndExit(e);
            }
            if (buf.length < 2 || buf[0] != '*') {
                logFormatErrAndExit(readLines);
            }
            // 取出命令参数，比如 *3\r\n 中的 3
            int argc = getLength(buf, 1);
            // 至少要有一个参数（被调用的命令）
            if (argc < 1) {
                logFormatErrAndExit(readLines);
            }
            fakeClient.setArgc(argc);
            for (int j = 0; j < argc; j++) {
                try {
                    line = br.readLine();
                    readLines++;
                    if (line == null) {
                        // 读完，跳出
                        break;
                    }
                    buf = line.toCharArray();
                } catch (IOException e) {
                    logReadErrAndExit(e);
                }
                if (buf.length < 2 || buf[0] != '$') {
                    logFormatErrAndExit(readLines);
                }
                // 读取参数值的长度
                int len = getLength(buf, 1);
                if (len == 0) {
                    // 先不允许出现 0
                    logFormatErrAndExit(readLines);
                }
                // 读取参数值 (编码的是字节长度而不是字符长度)
                byte[] cbuf = new byte[len];
                try {
                    // 不能 readline 是因为防止文本中出现
                    // 只有一个 `\n` 或 `\r` 的情况
                    br.read(cbuf);
                    // 把 CRLF 换行读掉
                    br.readShort();
                    readLines++;
                } catch (IOException e) {
                    logReadErrAndExit(e);
                }
                String val = new String(cbuf, StandardCharsets.UTF_8);
                fakeClient.addArgv(val);
            }
            Command cmd = sc.lookupCommand(fakeClient.argv(0));
            if (cmd == null) {
                logger.log(Level.WARNING, String.format("Unknown command '%s'  reading the append only file", fakeClient.argv(0)));
                System.exit(1);
            }
            //调用伪客户端，执行命令
            cmd.proc(fakeClient);
            // 清理命令和命令参数对象
            fakeClient.resetArgAndCmd();
        }
        // 如果能执行到这里，说明 AOF 文件的全部内容都可以正确地读取
        // 完成进度条
        progressBar.forceComplete();
        // 关闭 AOF 文件
        try {
            br.close();
        } catch (IOException e) {
            logger.log(Level.WARNING, String.format("Can't close the append log file: %s", e.getMessage()));
        }
        // 复原 AOF 状态
        aofAttr.setState(oldAofState);
        // 停止载入
        loadingInfo.stopLoading();
        // 更新服务器状态中， AOF 文件的当前大小
        updateAofCurrentSize();
        // 记录前一次重写时的大小 （还未实现重写功能）
        return KeyCrafter_OK;
    }
    private static void logFormatErrAndExit(int line) {
        logger.log(Level.WARNING, String.format("Bad file format reading the append only file: near line %d",line));
        System.exit(1);
    }

    private static void logReadErrAndExit(IOException e) {
        logger.log(Level.WARNING, String.format("An exception occurred while reading the append only file: %s", e.getMessage()));
        System.exit(1);
    }
    private static void formatErrLog() {
        logger.log(Level.WARNING, "Bad file format reading the append only file");
        System.exit(1);
    }

    /*将 aof 文件的当前大小记录到服务器状态中。
     * 通常用于 BGREWRITEAOF 执行之后，或者服务器重启之后。
     */
    static void updateAofCurrentSize() {
        // 读取文件状态
//        FileChannel channel = sc.getAOFAttribute().channel;
        // channel 已经关闭了？
    }

    private static int getLength(char[] buf, int start) {
        int i = start;
        if (buf[i] > '9' || buf[i] < '0') {
            throw new RuntimeException("Illegal number");
        }
        int len = 0;
        while (i < buf.length) {
            len = len * 10 + buf[i] - '0';
            i++;
        }
        return len;
    }

    // AOF 后台重写
    // 在进行 AOF 重写时，必须复制一份数据而不是直接引用原数据
    // 因为元数据在重写的过程中可能会被修改，导致重写的数据出错
    public static int rewriteAppendOnlyFileBackground() {
        // 在 java 中没有 fork（没找到类似功能的函数，windows 系统本身也不支持fork），
        // 这里用线程代替进程执行重写操作
        AofAttribute aofAttr = sc.getAOFAttribute();
        if (aofAttr.getRewriteThreadId() != -1) {
            // 已经有线程在进行 AOF 重写了
            return KeyCrafter_ERR;
        }
        // 记录数据复制开始前的时间，计算数据复制耗时用
        //long start = TimeUtil.microTime();
        KVDB[] dbs = dataCopy(sc);

        // 执行异步任务
        CompletableFuture<Integer> cf = CompletableFuture
                // 异步任务
                .supplyAsync(() -> {
                    long tid = Thread.currentThread().getId();
                    // 设置执行 AOF 重写的线程 id
                    aofAttr.setRewriteThreadId(tid);
                    Thread.currentThread().setName("KeyCrafter-aof-rewrite");
                    String tmpfile = String.format("temp-rewriteaof-bg-%d.aof", tid);
                    return  rewriteAppendOnlyFile(tmpfile, dbs);
                }, aofRewriteExecutor)
                //异常
                .exceptionally(e -> {
                    e.printStackTrace();
                    return KeyCrafter_ERR;
                });
        aofAttr.setRewriteFuture(cf);
        /*  将 aof_selected_db 设为 -1 ，
         * 强制让 feedAppendOnlyFile() 下次执行时引发一个 SELECT 命令，
         * 从而确保之后新添加的命令会设置到正确的数据库中
         */
        aofAttr.setSelectedDb(-1);
        return KeyCrafter_OK;
    }

    static int rewriteAppendOnlyFile(String filename,KVDB[] dbs) {
        long now = microTime();
        /*
         * 创建临时文件
         *
         */
        String tmpFile = String.format("temp-rewriteaof-%d.aof",Thread.currentThread().getId());
        Path path = Paths.get(sc.getWorkDir(), tmpFile);
        File file = path.toFile();
        BufferedOutputStream out;
        try {
            file.createNewFile();
            out = new BufferedOutputStream(Files.newOutputStream(file.toPath()));
        } catch (IOException e) {
            throw new RuntimeException(e);
        }

        // 遍历所有数据库
        for (int j = 0; j < dbs.length; j++){
            KVDB db = dbs[j];
            // 指向键空间
            Dict dict = db.getKeySpace();
            // 创建键空间迭代器
            DictIterator di = dict.getSafeIterator();
            try {
                if (di == null) {
                    out.close();
                    return KeyCrafter_ERR;
                }
                // 首先写入 SELECT 命令，确保之后的数据会被插入到正确的数据库上
                String[] selectCmd = {"select", String.valueOf(j)};
                writeArray(out, selectCmd);
                // 遍历数据库所有键，并通过命令将它们的当前状态（值）记录到新 AOF 文件中
                Dict.Entry de;
                while((de = di.next()) != null) {
                    // 取出键
                    String key = (String) de.getKey();
                    // 取出值
                    KeyCrafterObject o = (KeyCrafterObject) de.getValue();
                    // 取出过期时间
                    long expireTime = db.getExpire(key);
                    // 如果键已经过期，那么跳过它，不保存
                    if (expireTime != -1 || expireTime > now) {
                        continue;
                    }
                    // 根据值的类型，选择适当的命令来保存值
                    if (o.getType() == KeyCrafter_STRING) {
                        rewriteStringObject(out,key, (StringObj) o);
                    } else if (o.getType() == KeyCrafter_LIST) {
                        rewriteListObject(out, key, (ListObj) o);
                    } else if (o.getType() == KeyCrafter_HASH) {
                        rewriteHashObject(out, key, (HashObj) o);
                    } else if (o.getType() == KeyCrafter_SET) {
                        rewriteSetObject(out, key, (SetObj) o);
                    }
                }
            } catch (Exception e) {
                throw new RuntimeException(e);
            }
        }
        // 冲洗并关闭新 AOF 文件
        try {
            out.flush();
            out.close();
        } catch (IOException e) {
            logger.log(Level.WARNING,String.format("Write error writing append only file on disk: %s" ,e.getMessage()));
            return KeyCrafter_ERR;
        }
        /*  原子地改名，用重写后的新 AOF 文件覆盖旧 AOF 文件
         */
        if(!file.renameTo(new File(filename))) {
            logger.log(Level.WARNING,"Error moving temp append only file on the final destination");
            return KeyCrafter_ERR;
        }

        logger.log(Level.INFO,"SYNC append only file rewrite performed");

        return KeyCrafter_OK;
    }
    // 重新写入SADD
    private static void rewriteSetObject(BufferedOutputStream out, String key, SetObj value) throws IOException {
        SetType setType = value.getValue();
        int count = 0;
        Iterator iterator = setType.initIterator();
        Object ele;
        String[] cmd = new String[KeyCrafter_AOF_REWRITE_ITEMS_PER_CMD + 2];
        while ((ele = iterator.next()) != null){
            if (count == 0){
                cmd[count++] = "sadd";
                cmd[count++] = key;
            }
            // 填入值
            cmd[count++] = ele.toString();
            if (count == KeyCrafter_AOF_REWRITE_ITEMS_PER_CMD + 2){
                writeArray(out, cmd);
                count = 0;
            }
        }
    }

    // 重新写入HMSET
    private static void rewriteHashObject(BufferedOutputStream out, String key, HashObj value) throws IOException {
        HashType hashType = value.getValue();
        int count = 0; // 这里的 count 表示命令参数的总数
        Iterator<Map.Entry> iterator = hashType.iterator();
        Map.Entry entry;
        String[] cmd = new String[KeyCrafter_AOF_REWRITE_ITEMS_PER_CMD + 2];
        while ((entry = iterator.next()) != null){
            if (count == 0) {
                cmd[count++] = "hmset";
                cmd[count++] = key;
            }
            // 取出 field
            cmd[count++] = entry.getKey().toString();
            // 取出 value
            cmd[count++] = entry.getValue().toString();
            if (count == KeyCrafter_AOF_REWRITE_ITEMS_PER_CMD + 2) {
                writeArray(out, cmd);
                count = 0;
            }
        }
        // 写入不足一个数组的命令
        writeArray(out, cmd, 0, count);
    }

    // 重写 String 对象
    private static void rewriteStringObject(OutputStream out, String key, StringObj value) throws IOException {
        String[] cmd = {"set",key,value.getValue().getString()};
        writeArray(out, cmd);
    }
    /*  将重建列表对象所需的命令写入到 r 。
     * 命令的形式如下：  RPUSH item1 item2 ... itemN
     */
    private static void rewriteListObject(OutputStream out, String key,ListObj value) throws IOException {
        ListType listType = value.getValue();
        int count = 0;
        Iterator iterator = listType.iterator();
        String[] cmd = new String[KeyCrafter_AOF_REWRITE_ITEMS_PER_CMD + 2];
        while (iterator.hasNext()) {
            Object ele = iterator.next();
            if (count == 0) {
                cmd[count] = "rpush";
                cmd[count++] = key;
            }
            // 取出值
            cmd[count++] = ele.toString();
            // 元素计数
            if (++count == KeyCrafter_AOF_REWRITE_ITEMS_PER_CMD + 2) {
                writeArray(out, cmd);
                count = 0;
            }
        }
        // 写入不足一个数组的命令
        writeArray(out, cmd,0,count + 2);
    }

    private static KVDB[] dataCopy(ServerContext sc) {
        int n = sc.dbNum();
        KVDB[] copy = new KVDB[n];
        Kryo kryo = new Kryo();
        initKryo(kryo);
        for (int i = 0; i < n; i++) {
            KVDB dup = kryo.copy(sc.db(i));
            copy[i] = dup;
        }
        return copy;
    }

    // 引入序列化工具kryo
    private static void initKryo(Kryo kryo) {
        if (kryo == null) {
            throw new NullPointerException("Kryo is null");
        }
        kryo.setRegistrationRequired(false);

        DictType dictType = new MurmurHash2();
        // 注册类
        kryo.register(DefaultKVDB.class).setInstantiator(() -> KVDB.newDefaultDb(-1));
        kryo.register(MurmurHash2.class).setInstantiator(() -> dictType);

        kryo.register(HashDict.class).setInstantiator(() -> new HashDict(null, null));

        kryo.register(StringObj.class).setInstantiator(() -> new StringObj(null));
        kryo.register(HashObj.class).setInstantiator(() -> new HashObj(null));
        kryo.register(ListObj.class).setInstantiator(() -> new ListObj(null));
        kryo.register(SetObj.class).setInstantiator(() -> new SetObj(null));
        kryo.register(ZSetObj.class).setInstantiator(() -> new ZSetObj(null));

        kryo.register(DefaultStringType.class).setInstantiator(() -> new DefaultStringType(null));
    }

    /*
     * 当子线程完成 AOF 重写时，父线程调用这个函数。
     */
    static void backgroundRewriteDoneHandler(int exitCode) {
        AofAttribute aof = sc.getAOFAttribute();
        if (exitCode == KeyCrafter_OK) {
            logger.log(Level.INFO,"Background AOF rewrite terminated with success");
            // 打开保存新 AOF 文件内容的临时文件
            Path tmpPath = Paths.get(sc.getWorkDir(),
                    String.format("temp-rewriteaof-bg-%d.aof", aof.getRewriteThreadId()));
            FileChannel newChannel = null;
            try {
                // 设置操作选项
                Set<OpenOption> options = new HashSet<>();
                // 以追加模式写入
                options.add(StandardOpenOption.APPEND);
                if (aof.getFsync() == AOF_FSYNC_ALWAYS) {
                    // 如果是 FSYNC_ALWAYS (总是同步) 模式
                    options.add(StandardOpenOption.DSYNC);
                }
                // 打开临时文件 channel
                newChannel = FileChannel.open(tmpPath, options);
            } catch (IOException e) {
                logger.log(Level.WARNING,
                        String.format("Unable to open the temporary AOF produced by the child: %s",e.getMessage()));
            }
            // 将累积的重写缓存写入到临时文件中
            // 这个函数调用的 write 操作会阻塞线程
            int bytesNum = 0;
            try {
                bytesNum = aofRewriteBufferWrit(newChannel);
            } catch (IOException e) {
                logger.log(Level.WARNING,
                        String.format("Error trying to flush the parent diff to the rewritten AOF: %s",e.getMessage()));
            }
            logger.log(Level.INFO,
                    String.format("Parent diff successfully flushed to the rewritten AOF (%d bytes)",bytesNum));

            // 文件替换
            try {
                // 关闭原 AOF 文件
                aof.getChannel().close();
                // 删除原文件
                Path aofPath = Paths.get(aof.getAofFilename());
                Files.delete(aofPath);
                // 临时文件重命名
                tmpPath.toFile().renameTo(aofPath.toFile());
            } catch (IOException e) {
                throw new RuntimeException(e);
            }
            // 设置新的 channel
            aof.setChannel(newChannel);
            logger.log(Level.INFO, "Background AOF rewrite finished successfully");

            // 先在这里更改 id
            aof.setRewriteThreadId(-1);
        } else {
            logger.log(Level.WARNING,
                    "Background AOF rewrite terminated with error");
        }
    }

    // 写入重写重写缓冲区的内容
    // 返回写入写字节数
    static int aofRewriteBufferWrit(FileChannel channel) throws IOException {
        ByteBuf rewriteBuf = sc.getAOFAttribute().getRewriteBuf();
        int len = rewriteBuf.readableBytes();
        if (rewriteBuf.isReadable()) {
            rewriteBuf.readBytes(channel,len);
            rewriteBuf.clear();
        }
        return len;
    }
}
